package com.codi.bus.core.util;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 1、用ThreadPoolExecutor自定义线程池，看线程是的用途，如果任务量不大，可以用无界队列，如果任务量非常大，要用有界队列，防止OOM
 * 2、如果任务量很大，还要求每个任务都处理成功，要对提交的任务进行阻塞提交，重写拒绝机制，改为阻塞提交。保证不抛弃一个任务
 * 3、最大线程数一般设为2N+1最好，N是CPU核数
 * 4、核心线程数，看应用，如果是任务，一天跑一次，设置为0，合适，因为跑完就停掉了，如果是常用线程池，看任务量，是保留一个核心还是几个核心线程数
 * 5、如果要获取任务执行结果，用CompletionService，但是注意，获取任务的结果的要重新开一个线程获取，如果在主线程获取，就要等任务都提交后才获取，就会阻塞大量任务结果，队列过大OOM，所以最好异步开个线程获取结果
 * @作者：曾祥亮
 * @时间：2017年10月16日 上午10:12:59
 */
public class ThreadPoolExecutorUtil {
	private ThreadPoolExecutor pool = null;
	//核心线程池大小
	private int corePoolSize;
	//最大线程池大小
	private int maxPoolSize;
	//线程池中超过corePoolSize数目的空闲线程最大存活时间,单位：秒
	private long keepAliveTime;
	//阻塞队列大小
	private int blockSize;
	//创建线程池的类型:true=阻塞  false=非阻塞
	private boolean isBlock;
	/**
	 * @param corePoolSize 核心线程池大小
	 * @param maxPoolSize 最大线程池大小
	 * @param keepAliveTime 线程池中超过corePoolSize数目的空闲线程最大存活时间,单位：秒
	 * @param blockSize 阻塞队列大小
	 */
	public ThreadPoolExecutorUtil(int corePoolSize, int maxPoolSize, long keepAliveTime, int blockSize, boolean isBlock){
		this.corePoolSize = corePoolSize;
		this.maxPoolSize = maxPoolSize;
		this.keepAliveTime = keepAliveTime;
		this.blockSize = blockSize;
		this.isBlock = isBlock;
		init();
	}
	public ThreadPoolExecutorUtil(){
		init();
	}
	public ThreadPoolExecutorUtil(boolean isBlock){
		this.isBlock = isBlock;
		init();
	}
	/**
     * 线程池初始化方法
     *
     * corePoolSize 核心线程池大小----10
     * maxPoolSize 最大线程池大小----30
     * keepAliveTime 线程池中超过corePoolSize数目的空闲线程最大存活时间----30+单位TimeUnit
     * TimeUnit keepAliveTime时间单位----TimeUnit.MINUTES
     * workQueue 阻塞队列----new ArrayBlockingQueue<Runnable>(10)====10容量的阻塞队列
     * threadFactory 新建线程工厂----new CustomThreadFactory()====定制的线程工厂
     * rejectedExecutionHandler 当提交任务数超过maxmumPoolSize+workQueue之和时,
     *                          即当提交第41个任务时(前面线程都没有执行完,此测试方法中用sleep(100)),
     *                                任务会交给RejectedExecutionHandler来处理
     */
	public void init() {
		if(corePoolSize <=0){
			corePoolSize=1;
		}
		if(maxPoolSize <=0){
			maxPoolSize=3;
		}
		if(keepAliveTime <=0){
			keepAliveTime=5;
		}
		if(blockSize <=0){
			blockSize=5;
		}
		if(isBlock){
			pool = new ThreadPoolExecutor(
	        		corePoolSize,
	        		maxPoolSize,
	        		keepAliveTime,
	                TimeUnit.SECONDS,
	                new ArrayBlockingQueue<Runnable>(blockSize),
	                new CustomThreadFactory(),
	                new BlockRejectedExecutionHandler());
		}else{
			pool = new ThreadPoolExecutor(
	        		corePoolSize,
	        		maxPoolSize,
	        		keepAliveTime,
	                TimeUnit.SECONDS,
	                new ArrayBlockingQueue<Runnable>(blockSize),
	                new CustomThreadFactory(),
	                new NoBlockRejectedExecutionHandler());
		}

    }
	public void destory() {
        if(pool != null) {
            pool.shutdownNow();
        }
    }
	public ExecutorService getThreadPoolExecutor() {
        return this.pool;
    }
	private class CustomThreadFactory implements ThreadFactory {
        private AtomicInteger count = new AtomicInteger(0);
        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r);
            String threadName = ThreadPoolExecutorUtil.class.getSimpleName() + count.addAndGet(1);
            t.setName(threadName);
            return t;
        }
    }
	//非阻塞
	private class NoBlockRejectedExecutionHandler implements RejectedExecutionHandler {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            // 记录异常
            // 报警处理等
            System.out.println("线程异常:"+ ThreadPoolExecutorUtil.class.getSimpleName());
        }
    }
	//阻塞
	private class BlockRejectedExecutionHandler implements RejectedExecutionHandler {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            try {
                // 核心改造点，由blockingqueue的offer改成put阻塞方法
                executor.getQueue().put(r);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
